package ooo.reindeer.concurrent.netqueue;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import ooo.reindeer.concurrent.netqueue.handler.QueueClientMessageHandler;
import ooo.reindeer.concurrent.netqueue.handler.QueueServerMessageHandler;
import ooo.reindeer.concurrent.netqueue.message.Command;
import ooo.reindeer.concurrent.netqueue.message.Message;
import ooo.reindeer.concurrent.netqueue.message.MessageCoder;
import ooo.reindeer.logging.ILogger;
import ooo.reindeer.logging.LogUtil;

import java.nio.ByteOrder;

public class NetQueue {


    private static final ILogger logger = LogUtil.getLogger(NetQueue.class);
    ServerBootstrap serverBootstrap;
    Bootstrap bootstrap;
    NioEventLoopGroup boos;
    NioEventLoopGroup worker;
    QueueServerMessageHandler handler;
    ConnectListener connectListener;

    public static NetQueue newClient(String host, int port, boolean reconnect) throws InterruptedException {
        NetQueue netQueue = new NetQueue();
        netQueue.bootstrap = new Bootstrap();
        netQueue.boos = new NioEventLoopGroup();
        netQueue.handler = new QueueServerMessageHandler();
        final QueueServerMessageHandler finalHandler = netQueue.handler;
        netQueue.bootstrap.group(netQueue.boos)
                .channel(NioSocketChannel.class).handler(
                new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
//                        nioSocketChannel.pipeline().addLast(new LoggingHandler(LogLevel.WARN));
                        nioSocketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN, 16 * 1024, 0, 4, 0, 0, false));
                        nioSocketChannel.pipeline().addLast(new MessageCoder());
                        nioSocketChannel.pipeline().addLast(finalHandler);
                    }
                }
        );

        netQueue.connectListener = new ConnectListener(reconnect);

        netQueue.connectListener.watch(netQueue.bootstrap, host, port);


        return netQueue;
    }

    public static void newServer(String host, int port) throws InterruptedException {
        NetQueue netQueue = new NetQueue();
        netQueue.serverBootstrap = new ServerBootstrap();
        netQueue.boos = new NioEventLoopGroup();
        netQueue.worker = new NioEventLoopGroup();
        netQueue.serverBootstrap
                .group(netQueue.boos, netQueue.worker).channel(NioServerSocketChannel.class).childHandler(
                new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
//                        nioSocketChannel.pipeline().addLast(new LoggingHandler(LogLevel.WARN));
                        nioSocketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN, 16 * 1024, 0, 4, 0, 0, false));
                        nioSocketChannel.pipeline().addLast(new MessageCoder());
                        nioSocketChannel.pipeline().addLast(new QueueClientMessageHandler());
                    }
                }
        );
        netQueue.serverBootstrap.bind(host, port).sync();
    }

    public void cancelRegiste(String topic, IMessageHandler handler) {
        this.handler.cancelRegiste(topic);
        Message registe = new Message();
        registe.setCommand(Command.CANCEL_REGISTE);
        registe.setTopic(topic);
        try {
            this.handler.cancelRegiste(topic);
            connectListener.getChannelFuture().channel().writeAndFlush(registe).sync();
        } catch (Exception e) {
            logger.warn("Connection exception cancel registration event will be delayed", e.getMessage());
        }

    }

    public void put(String topic, byte[] data) {
        Message put = new Message();
        put.setCommand(Command.PUT);
        put.setTopic(topic);
        put.setData(data);
        connectListener.getChannelFuture().channel().writeAndFlush(put);
    }

    public void registe(String topic, IMessageHandler handler) {

        Message registe = new Message();
        registe.setCommand(Command.REGISTE);
        registe.setTopic(topic);

        try {
            this.handler.registe(topic, handler);
            connectListener.getChannelFuture().channel().writeAndFlush(registe).sync();
        } catch (Exception e) {
            logger.warn("Connection exception registration event will be delayed", e.getMessage());
        }


    }

}
